package com.ztf666.service.impl;

import com.ztf666.service.QueryService;
import com.ztf666.utils.ConnectionUtils;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.search.*;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.ml.job.config.Operator;
import org.elasticsearch.common.text.Text;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.aggregations.AggregationBuilder;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.range.ParsedDateRange;
import org.elasticsearch.search.aggregations.bucket.range.ParsedRange;
import org.elasticsearch.search.aggregations.metrics.ParsedCardinality;
import org.elasticsearch.search.aggregations.metrics.ParsedExtendedStats;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightField;

import java.io.IOException;
import java.util.Map;

/**
 * @Author ztf
 * @Date 21:47 2022/1/17
 **/
public class QueryServiceImpl implements QueryService {
    @Override
    public void termQuery(String index, String filed, String value) throws Exception {
        SearchResponse response = baseSearch(index, QueryBuilders.termQuery(filed, value));
        System.err.println(response.toString());
    }

    @Override
    public void termsQuery(String index, String filed, String... value) throws Exception {
        SearchResponse response = baseSearch(index, QueryBuilders.termsQuery(filed, value));
        System.err.println(response.toString());
    }

    @Override
    public void matchAll(String index) throws Exception {
        SearchResponse response = baseSearch(index, QueryBuilders.matchAllQuery());
        System.err.println(response.toString());
    }

    @Override
    public void match(String index, String field, String value) throws Exception {
        SearchResponse response = baseSearch(index, QueryBuilders.matchQuery(field, value));
        System.err.println(response.toString());
    }

    @Override
    public void boolMatch(String index, String field, String value, org.elasticsearch.index.query.Operator operator) throws Exception {
        SearchResponse response = baseSearch(index, QueryBuilders.matchQuery(field, value).operator(operator));
        System.err.println(response.toString());
    }



    @Override
    public void multiMacth(String index, String value, String... fields) throws Exception {
        SearchResponse response = baseSearch(index, QueryBuilders.multiMatchQuery(value, fields));
        System.err.println(response.toString());
    }

    @Override
    public void idQuery(String index, String id) throws Exception {
        RestHighLevelClient client = ConnectionUtils.getClient();
        GetRequest getRequest = new GetRequest(index, id);
        GetResponse response = client.get(getRequest, RequestOptions.DEFAULT);
        System.err.println(response.toString());
    }

    @Override
    public void idsQuery(String index, String... ids) throws Exception {
        SearchResponse response = baseSearch(index, QueryBuilders.idsQuery().addIds(ids));
        System.err.println(response.toString());
    }

    @Override
    public void prefixQuery(String index, String field, String prefix) throws Exception {
        SearchResponse response = baseSearch(index, QueryBuilders.prefixQuery(field,prefix));
        System.err.println(response.toString());
    }

    @Override
    public void fuzzyQuery(String index, String field, String value, int prefix_length) throws Exception {
        SearchResponse response = baseSearch(index, QueryBuilders.fuzzyQuery(field,value).prefixLength(prefix_length));
        System.err.println(response.toString());
    }

    @Override
    public void wildcardQuery(String index, String field, String value) throws Exception {
        SearchResponse response = baseSearch(index, QueryBuilders.wildcardQuery(field,value));
        System.err.println(response.toString());
    }

    @Override
    public void rangeQuery(String index,String field,long gte,long lte) throws Exception {
        SearchResponse response = baseSearch(index, QueryBuilders.rangeQuery(field).gte(gte).lte(lte));
        System.err.println(response.toString());
    }

    @Override
    public void regQuery(String index, String field, String reg) throws Exception {
        SearchResponse response = baseSearch(index,QueryBuilders.regexpQuery(field,reg));
        System.err.println(response.toString());
    }

    @Override
    public String createScrollQuery(String index, QueryBuilder queryBuilder, int size) throws Exception {
        RestHighLevelClient client = ConnectionUtils.getClient();
        SearchRequest searchRequest = new SearchRequest(index);//查询指定的库
        searchRequest.scroll("1m");//开启滚动查询,时间为1分钟
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.size(size);//设置一次返回数据的长度
        searchSourceBuilder.query(queryBuilder);
        searchRequest.source(searchSourceBuilder);//设置一个搜索的条件
        SearchResponse response = client.search(searchRequest, RequestOptions.DEFAULT);
        String scrollId = response.getScrollId();
        System.err.println(response.toString());
        return scrollId;
    }

    @Override
    public void srollQuery( String scrollId) throws Exception {
        RestHighLevelClient client = ConnectionUtils.getClient();
        SearchScrollRequest searchRequest = new SearchScrollRequest(scrollId);
        SearchResponse response = client.scroll(searchRequest, RequestOptions.DEFAULT);
        System.err.println(response.toString());
    }

    @Override
    public void clearScroll(String index, String scrollId) throws Exception {
        RestHighLevelClient client = ConnectionUtils.getClient();
        ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
        clearScrollRequest.addScrollId(scrollId);//设置要清理的id
        ClearScrollResponse response = client.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);//清理滚动查询
    }

    @Override
    public void deleteByQuery(String index, QueryBuilder queryBuilder) throws Exception {
        RestHighLevelClient client = ConnectionUtils.getClient();
        DeleteByQueryRequest deleteByQueryRequest = new DeleteByQueryRequest(index);
        deleteByQueryRequest.setQuery(queryBuilder);//设置查询条件
        BulkByScrollResponse response = client.deleteByQuery(deleteByQueryRequest, RequestOptions.DEFAULT);//根据查询条件删除数据
        System.err.println(response.toString());
    }

    @Override
    public void boostQuery(String index, QueryBuilder queryBuilder, QueryBuilder boostBuilder,float boost) throws Exception {
        RestHighLevelClient client = ConnectionUtils.getClient();
        SearchRequest searchRequest = new SearchRequest(index);//查询指定的库
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        //通过builder来构建我们的查询条件以及权重的调整的条件
        searchSourceBuilder.query(QueryBuilders.boostingQuery(queryBuilder,boostBuilder).negativeBoost(boost));
        searchRequest.source(searchSourceBuilder);//设置一个搜索的条件
        SearchResponse response = client.search(searchRequest, RequestOptions.DEFAULT);
        System.err.println(response.toString());
    }

    @Override
    public void highLight(String index, QueryBuilder queryBuilder, HighlightBuilder highlightBuilder) throws Exception {
        RestHighLevelClient client = ConnectionUtils.getClient();
        SearchRequest searchRequest = new SearchRequest(index);
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        //构建的查询条件
        searchSourceBuilder.query(queryBuilder);
        searchSourceBuilder.highlighter(highlightBuilder);
        searchRequest.source(searchSourceBuilder);
        SearchResponse response = client.search(searchRequest, RequestOptions.DEFAULT);
        //先获取原始数据
        SearchHit[] hits = response.getHits().getHits();
        for (SearchHit hit : hits) {
            String id = hit.getId();
            Map<String, Object> hitSourceAsMap = hit.getSourceAsMap();
            Map<String, HighlightField> highlightFields = hit.getHighlightFields();

            highlightFields.forEach((fieldName,highlightField)->{
                if (highlightField != null) {
                    Text[] fragments = highlightField.getFragments();
                    if (fragments != null && fragments.length > 0) {
                        Text fragment = fragments[0];
                        //替换
                        //将高亮的数据替换掉原始map中的数据
                        hitSourceAsMap.put(fieldName, fragment.toString());
                    }
                }
            });

            System.err.println(hitSourceAsMap);
        }

    }

    @Override
    public void cardinalityAgg(String index, String field) throws Exception {
        SearchResponse response = baseAggre(index, AggregationBuilders.cardinality("bieming").field(field));
        Aggregations aggregations = response.getAggregations();
        ParsedCardinality aggregation = aggregations.get("bieming");
        System.err.println(aggregation.getValue());
    }


    @Override
    public void rangeAgg(String index, AggregationBuilder aggregationBuilder) throws Exception {
        SearchResponse response = baseAggre(index, aggregationBuilder);
        ParsedRange aggregation = response.getAggregations().get("bieming");
        aggregation.getBuckets().stream().forEach(value->System.out.println(value.getKey()+"--->"+value.getDocCount()));
    }

    @Override
    public void dateRangeAgg(String index, AggregationBuilder aggregationBuilder) throws Exception {
        SearchResponse response = baseAggre(index, aggregationBuilder);
        ParsedDateRange aggregation = response.getAggregations().get("bieming");
        aggregation.getBuckets().parallelStream().forEach(value->System.out.println(value.getKey()+"--->"+value.getDocCount()));
    }

    @Override
    public void ipRangeAgg(String index, AggregationBuilder aggregationBuilder) throws Exception {
        SearchResponse response = baseAggre(index, aggregationBuilder);
        System.err.println(response.toString());
    }

    @Override
    public void extendedStats(String index, String field) throws Exception {
        SearchResponse response = baseAggre(index, AggregationBuilders.extendedStats("bieming").field(field));
        ParsedExtendedStats aggregation = response.getAggregations().get("bieming");
        System.err.println(aggregation.getAvgAsString());
        System.err.println(response.toString());
    }


    private SearchResponse baseAggre(String index,AggregationBuilder aggregationBuilder) throws IOException {
        RestHighLevelClient client = ConnectionUtils.getClient();
        SearchRequest searchRequest = new SearchRequest(index);
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.aggregation(aggregationBuilder);
        searchRequest.source(searchSourceBuilder);
        SearchResponse response = client.search(searchRequest, RequestOptions.DEFAULT);
        return response;
    }




    public SearchResponse baseSearch(String index, QueryBuilder queryBuilder) throws Exception {
        RestHighLevelClient client = ConnectionUtils.getClient();
        SearchRequest searchRequest = new SearchRequest(index);
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        //通过builder来构建我们的查询条件
        searchSourceBuilder.query(queryBuilder);
        //设置一个搜索的条件
        searchRequest.source(searchSourceBuilder);
        SearchResponse response = client.search(searchRequest, RequestOptions.DEFAULT);
        return response;
    }

}
